<!DOCTYPE html>
<html lang="zh-cn">
<head>
   
    <link type="text/css" rel="stylesheet" href="/bundles/blog-common.css?v=KOZafwuaDasEedEenI5aTy8aXH0epbm6VUJ0v3vsT_Q1"/>
<link id="MainCss" type="text/css" rel="stylesheet" href="/skins/ThinkInside/bundle-ThinkInside.css?v=RRjf6pEarGnbXZ86qxNycPfQivwSKWRa4heYLB15rVE1"/>
<link type="text/css" rel="stylesheet" href="/blog/customcss/428549.css?v=%2fam3bBTkW5NBWhBE%2fD0lcyJv5UM%3d"/>

</head>
<body>
<a name="top"></a>

<div id="page_begin_html"></div><script>load_page_begin_html();</script>

<div id="topics">
	<div class = "post">
		<h1 class = "postTitle">
			<a id="cb_post_title_url" class="postTitle2" href="https://www.cnblogs.com/frankdeng/p/9255935.html">Hadoop案例（二）压缩解压缩</a>
		</h1>
		<div class="clear"></div>
		<div class="postBody">
			<div id="cnblogs_post_body" class="blogpost-body"><h2><strong>压缩</strong><strong>/</strong><strong>解压缩案例</strong></h2>
<h2><strong>一.&nbsp;</strong><strong><span style="font-family: 宋体;">对数据流的压缩和解压缩</span></strong></h2>
<p>CompressionCodec<span style="font-family: 宋体;">有两个方法可以用于轻松地压缩或解压缩数据。要想对正在被写入一个输出流的数据进行压缩，我们可以使用</span><span style="font-family: 'Times New Roman';">createOutputStream(OutputStreamout)</span><span style="font-family: 宋体;">方法创建一个</span><span style="font-family: 'Times New Roman';">CompressionOutputStream</span><span style="font-family: 宋体;">，将其以压缩格式写入底层的流。相反，要想对从输入流读取而来的数据进行解压缩，则调用</span><span style="font-family: 'Times New Roman';">createInputStream(InputStreamin)</span><span style="font-family: 宋体;">函数，从而获得一个</span><span style="font-family: 'Times New Roman';">CompressionInputStream</span><span style="font-family: 宋体;">，从而从底层的流读取未压缩的数据。</span></p>
<p>测试<span style="font-family: 宋体;">一下如下压缩方式</span>：</p>
<table border="1" cellspacing="0">
<tbody>
<tr>
<td valign="center" width="255">
<p>DEFLATE</p>
</td>
<td valign="center" width="293">
<p>org.apache.hadoop.io.compress.DefaultCodec</p>
</td>
</tr>
<tr>
<td valign="center" width="255">
<p>gzip</p>
</td>
<td valign="center" width="293">
<p>org.apache.hadoop.io.compress.GzipCodec</p>
</td>
</tr>
<tr>
<td valign="center" width="255">
<p>bzip2</p>
</td>
<td valign="center" width="293">
<p>org.apache.hadoop.io.compress.BZip2Codec</p>
</td>
</tr>
</tbody>
</table>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.mapreduce.compress;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> TestCompress {
    
    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) throws Exception, IOException {
</span><span style="color: #008000;">//</span><span style="color: #008000;">        compress("e:/test.txt","org.apache.hadoop.io.compress.BZip2Codec");</span>
        decompres(<span style="color: #800000;">"</span><span style="color: #800000;">e:/test.txt.bz2</span><span style="color: #800000;">"</span><span style="color: #000000;">);
    }
    
    </span><span style="color: #008000;">/*</span><span style="color: #008000;">
     * 压缩
     * filername：要压缩文件的路径
     * method：欲使用的压缩的方法（org.apache.hadoop.io.compress.BZip2Codec）
     </span><span style="color: #008000;">*/</span>
    <span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> compress(String filername, String method) throws ClassNotFoundException, IOException {
        
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 1 创建压缩文件路径的输入流</span>
        File fileIn = <span style="color: #0000ff;">new</span><span style="color: #000000;"> File(filername);
        InputStream </span><span style="color: #0000ff;">in</span> = <span style="color: #0000ff;">new</span><span style="color: #000000;"> FileInputStream(fileIn);
        
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 2 获取压缩的方式的类</span>
        Class codecClass =<span style="color: #000000;"> Class.forName(method);
        
        Configuration conf </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Configuration();
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 3 通过名称找到对应的编码/解码器</span>
        CompressionCodec codec =<span style="color: #000000;"> (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);

        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 4 该压缩方法对应的文件扩展名</span>
        File fileOut = <span style="color: #0000ff;">new</span> File(filername +<span style="color: #000000;"> codec.getDefaultExtension());

        OutputStream </span><span style="color: #0000ff;">out</span> = <span style="color: #0000ff;">new</span><span style="color: #000000;"> FileOutputStream(fileOut);
        CompressionOutputStream cout </span>= codec.createOutputStream(<span style="color: #0000ff;">out</span><span style="color: #000000;">);

        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 5 流对接</span>
        IOUtils.copyBytes(<span style="color: #0000ff;">in</span>, cout, <span style="color: #800080;">1024</span> * <span style="color: #800080;">1024</span> * <span style="color: #800080;">5</span>, <span style="color: #0000ff;">false</span>); <span style="color: #008000;">//</span><span style="color: #008000;"> 缓冲区设为5MB

        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 6 关闭资源</span>
        <span style="color: #0000ff;">in</span><span style="color: #000000;">.close();
        cout.close();
        </span><span style="color: #0000ff;">out</span><span style="color: #000000;">.close();
    }

    </span><span style="color: #008000;">/*</span><span style="color: #008000;">
     * 解压缩
     * filename：希望解压的文件路径
     </span><span style="color: #008000;">*/</span>
    <span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> decompres(String filename) throws FileNotFoundException, IOException {

        Configuration conf </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Configuration();
        CompressionCodecFactory factory </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> CompressionCodecFactory(conf);
        
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 1 获取文件的压缩方法</span>
        CompressionCodec codec = factory.getCodec(<span style="color: #0000ff;">new</span><span style="color: #000000;"> Path(filename));
        
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 2 判断该压缩方法是否存在</span>
        <span style="color: #0000ff;">if</span> (<span style="color: #0000ff;">null</span> ==<span style="color: #000000;"> codec) {
            System.</span><span style="color: #0000ff;">out</span>.println(<span style="color: #800000;">"</span><span style="color: #800000;">Cannot find codec for file </span><span style="color: #800000;">"</span> +<span style="color: #000000;"> filename);
            </span><span style="color: #0000ff;">return</span><span style="color: #000000;">;
        }

        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 3 创建压缩文件的输入流</span>
        InputStream cin = codec.createInputStream(<span style="color: #0000ff;">new</span><span style="color: #000000;"> FileInputStream(filename));
        
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 4 创建解压缩文件的输出流</span>
        File fout = <span style="color: #0000ff;">new</span> File(filename + <span style="color: #800000;">"</span><span style="color: #800000;">.decoded</span><span style="color: #800000;">"</span><span style="color: #000000;">);
        OutputStream </span><span style="color: #0000ff;">out</span> = <span style="color: #0000ff;">new</span><span style="color: #000000;"> FileOutputStream(fout);

        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 5 流对接</span>
        IOUtils.copyBytes(cin, <span style="color: #0000ff;">out</span>, <span style="color: #800080;">1024</span> * <span style="color: #800080;">1024</span> * <span style="color: #800080;">5</span>, <span style="color: #0000ff;">false</span><span style="color: #000000;">);

        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 6 关闭资源</span>
<span style="color: #000000;">        cin.close();
        </span><span style="color: #0000ff;">out</span><span style="color: #000000;">.close();
    }
}</span></pre>
</div>
<h2><strong>二.&nbsp;</strong><strong><span style="font-family: 宋体;">在</span>Map<span style="font-family: 宋体;">输出</span></strong><strong><span style="font-family: 宋体;">端采用压缩</span></strong></h2>
<p><span style="font-family: 宋体;">即使你的</span>MapReduce<span style="font-family: 宋体;">的输入输出文件都是未压缩的文件，你仍然可以对</span><span style="font-family: 'Times New Roman';">map</span><span style="font-family: 宋体;">任务的中间结果输出做压缩，因为它要写在硬盘并且通过网络传输到</span><span style="font-family: 'Times New Roman';">reduce</span><span style="font-family: 宋体;">节点，对其压缩可以提高很多性能，这些工作只要设置两个属性即可，我们来看下代码怎么设置：</span></p>
<p>给大家<span style="font-family: 宋体;">提供的</span>hadoop<span style="font-family: 宋体;">源码支持的压缩格式有：</span>BZip2Codec 、DefaultCodec</p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> WordCountDriver {

    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration configuration </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Configuration();

        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 开启map端输出压缩</span>
        configuration.setBoolean(<span style="color: #800000;">"</span><span style="color: #800000;">mapreduce.map.output.compress</span><span style="color: #800000;">"</span>, <span style="color: #0000ff;">true</span><span style="color: #000000;">);
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 设置map端输出压缩方式</span>
        configuration.setClass(<span style="color: #800000;">"</span><span style="color: #800000;">mapreduce.map.output.compress.codec</span><span style="color: #800000;">"</span>, BZip2Codec.<span style="color: #0000ff;">class</span>, CompressionCodec.<span style="color: #0000ff;">class</span><span style="color: #000000;">);

        Job job </span>=<span style="color: #000000;"> Job.getInstance(configuration);

        job.setJarByClass(WordCountDriver.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);

        job.setMapperClass(WordCountMapper.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        job.setReducerClass(WordCountReducer.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);

        job.setMapOutputKeyClass(Text.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        job.setMapOutputValueClass(IntWritable.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);

        job.setOutputKeyClass(Text.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        job.setOutputValueClass(IntWritable.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);

        FileInputFormat.setInputPaths(job, </span><span style="color: #0000ff;">new</span> Path(args[<span style="color: #800080;">0</span><span style="color: #000000;">]));
        FileOutputFormat.setOutputPath(job, </span><span style="color: #0000ff;">new</span> Path(args[<span style="color: #800080;">1</span><span style="color: #000000;">]));

        boolean result </span>= job.waitForCompletion(<span style="color: #0000ff;">true</span><span style="color: #000000;">);

        System.exit(result </span>? <span style="color: #800080;">1</span> : <span style="color: #800080;">0</span><span style="color: #000000;">);
    }
}</span></pre>
</div>
<p>2<span style="font-family: 宋体;">）</span>Mapper<span style="font-family: 宋体;">保持</span><span style="font-family: 宋体;">不变</span></p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span> WordCountMapper extends Mapper&lt;LongWritable, Text, Text, IntWritable&gt;<span style="color: #000000;">{
    
    @Override
    </span><span style="color: #0000ff;">protected</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        
        String line </span>=<span style="color: #000000;"> value.toString();
        
        String[] words </span>= line.split(<span style="color: #800000;">"</span> <span style="color: #800000;">"</span><span style="color: #000000;">);
        
        </span><span style="color: #0000ff;">for</span><span style="color: #000000;">(String word:words){
            context.write(</span><span style="color: #0000ff;">new</span> Text(word), <span style="color: #0000ff;">new</span> IntWritable(<span style="color: #800080;">1</span><span style="color: #000000;">));
        }
    }
}</span></pre>
</div>
<p>3）Reducer保持<span style="font-family: 宋体;">不变</span></p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span> WordCountReducer extends Reducer&lt;Text, IntWritable, Text, IntWritable&gt;<span style="color: #000000;">{
    
    @Override
    </span><span style="color: #0000ff;">protected</span> <span style="color: #0000ff;">void</span> reduce(Text key, Iterable&lt;IntWritable&gt;<span style="color: #000000;"> values,
            Context context) throws IOException, InterruptedException {
        
        </span><span style="color: #0000ff;">int</span> count = <span style="color: #800080;">0</span><span style="color: #000000;">;
        
        </span><span style="color: #0000ff;">for</span><span style="color: #000000;">(IntWritable value:values){
            count </span>+= value.<span style="color: #0000ff;">get</span><span style="color: #000000;">();
        }
        
        context.write(key, </span><span style="color: #0000ff;">new</span><span style="color: #000000;"> IntWritable(count));
    }
}</span></pre>
</div>
<h2><strong>三.&nbsp;</strong><strong>在</strong><strong>Reduce</strong><strong>输出</strong><strong><span style="font-family: 宋体;">端采用压缩</span></strong></h2>
<p><span style="font-family: 宋体;">基于</span>workcount<span style="font-family: 宋体;">案例处理</span></p>
<p>1<span style="font-family: 宋体;">）</span><span style="font-family: 宋体;">修改驱动</span></p>
<div class="cnblogs_code">
<pre><span style="color: #000000;">package com.xyg.mapreduce.compress;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.compress.Lz4Codec;
import org.apache.hadoop.io.compress.SnappyCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> WordCountDriver {

    </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        
        Configuration configuration </span>= <span style="color: #0000ff;">new</span><span style="color: #000000;"> Configuration();
        
        Job job </span>=<span style="color: #000000;"> Job.getInstance(configuration);
        
        job.setJarByClass(WordCountDriver.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        
        job.setMapperClass(WordCountMapper.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        job.setReducerClass(WordCountReducer.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        
        job.setMapOutputKeyClass(Text.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        job.setMapOutputValueClass(IntWritable.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        
        job.setOutputKeyClass(Text.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        job.setOutputValueClass(IntWritable.</span><span style="color: #0000ff;">class</span><span style="color: #000000;">);
        
        FileInputFormat.setInputPaths(job, </span><span style="color: #0000ff;">new</span> Path(args[<span style="color: #800080;">0</span><span style="color: #000000;">]));
        FileOutputFormat.setOutputPath(job, </span><span style="color: #0000ff;">new</span> Path(args[<span style="color: #800080;">1</span><span style="color: #000000;">]));
        
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 设置reduce端输出压缩开启</span>
        FileOutputFormat.setCompressOutput(job, <span style="color: #0000ff;">true</span><span style="color: #000000;">);
        
        </span><span style="color: #008000;">//</span><span style="color: #008000;"> 设置压缩的方式</span>
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.<span style="color: #0000ff;">class</span><span style="color: #000000;">); 
</span><span style="color: #008000;">//</span><span style="color: #008000;">        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); 
</span><span style="color: #008000;">//</span><span style="color: #008000;">        FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class); </span>
<span style="color: #000000;">        
        boolean result </span>= job.waitForCompletion(<span style="color: #0000ff;">true</span><span style="color: #000000;">);
        
        System.exit(result</span>?<span style="color: #800080;">1</span>:<span style="color: #800080;">0</span><span style="color: #000000;">);
    }
}</span></pre>
</div>
<p>2<span style="font-family: 宋体;">）</span>Mapper<span style="font-family: 宋体;">和</span>Reducer保持<span style="font-family: 宋体;">不变</span></p></div>

</body>
</html>
